package com.jasongj.kafka.consumer;

import java.io.*;
import java.security.MessageDigest;
import java.text.SimpleDateFormat;
import java.util.*;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import com.jasongj.kafka.consumer.GetFileSHA512;

import static com.jasongj.kafka.consumer.GetFileSHA512.getFileSHA512;

public class WriteBlock {
    private static final Logger logger = LogManager.getLogger(WriteBlock.class);
    public static void main(String[] args) throws Exception {
        //args = new String[] { "kafka0:9092", "topic1", "group2", "consumer2" };
        args = new String[]{"localhost:9092", "topic1", "test-consumer-group", "consumer3"};
        if (args == null || args.length != 4) {
            System.err.println(
                    "Usage:\n\tjava -jar kafka_consumer.jar ${bootstrap_server} ${topic_name} ${group_name} ${client_id}");
            System.exit(1);
        }
        String bootstrap = args[0];
        String topic = args[1];
        String groupid = args[2];
        String clientid = args[3];

        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrap);
        props.put("group.id", groupid);
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("max.poll.interval.ms", "300000");
        props.put("max.poll.records", "500");
        props.put("auto.offset.reset", "earliest");
        props.put("deserializer.encoding", "UTF-8");
        org.apache.kafka.clients.consumer.KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));
        File merklerFile = new File("d:\\log_dir\\merkler");
        if (!merklerFile.exists()) {
            merklerFile.createNewFile();
        }
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            records.partitions().forEach(topicPartition -> {
                List<ConsumerRecord<String, String>> partitionRecords = records.records(topicPartition);
                partitionRecords.forEach(record -> {
                    String temp = record.value()+"\n\r";
                    //System.out.println(temp);
                    try {
                        writeFile( temp);
                    } catch (IOException e) {
                        logger.info(e.getMessage());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    temp = "";
                });
                long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                consumer.commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(lastOffset + 1)));
            });
        }

    }
    static File merklerFile = null;
    static List<String> signatures = new ArrayList<String>();
    public static synchronized  void writeFile( String content) throws Exception {
        Date now = new Date();
        SimpleDateFormat dateFormatHour= new SimpleDateFormat("yyyy-MM-dd-HH");
        String  hour = dateFormatHour.format( now );
        SimpleDateFormat dateFormatMinute= new SimpleDateFormat("yyyy-MM-dd-HH-mm");
        String  minute = dateFormatMinute.format( now );

        //File dir = new File("/home/ubuntu/writeBlock/"+hour+"/");
        //File file = new File("/home/ubuntu/writeBlock/"+hour+"/"+minute);
        File dir = new File("d:\\log_dir\\"+hour+"\\");
        File file = new File("d:\\log_dir\\"+hour+"\\"+minute);

        //判断日期文件夹是否存在，不存在的话创建
        if (!dir.isDirectory()) {
            dir.mkdir();
        }
        //判断文件是否存在，不存在的话创建
        if (!file.exists()) {
            file.createNewFile();
        }
        BufferedWriter out = new BufferedWriter(new OutputStreamWriter(
                    new FileOutputStream(file, true)));
        out.write(content);
        out.close();
       // writeMerkle();

    }
    public static void writeMerkle() throws Exception {
        Date now = new Date();
        long time = 2*60*1000;//2分钟
        Date beforeDate = new Date(now .getTime() - time);//2分钟前的时间
        SimpleDateFormat dateFormatMinute= new SimpleDateFormat("yyyy-MM-dd-HH-mm");
        String  minute = dateFormatMinute.format( beforeDate );
        SimpleDateFormat dateFormatHour= new SimpleDateFormat("yyyy-MM-dd-HH");
        String  hour = dateFormatHour.format( beforeDate );

        File file = new File("d:\\log_dir\\"+hour+"\\"+minute);
        if (file.exists()) {
            String hashCode = getFileSHA512(file);
            System.out.println(merklerFile.getName() +" hash code is:"+hashCode);
            signatures.add(hashCode);
            if(signatures.size()==2){
                MerkleTree leaf = new MerkleTree(signatures);
                byte[] serializedTree = leaf.serialize();
                writeFileByBytes("d:\\log_dir\\merkler",serializedTree,true);
                signatures.clear();
            }
        }

    }
    public static boolean checkFileWritingOn(File file) throws Exception{
        long oldLen = 0;
        long newLen = 0;
        //File file = new File(fileName);
        while(true){
            newLen = file.length();
            if ((newLen - oldLen) > 0) {
                oldLen = newLen;
                System.out.println(file.length());
                Thread.sleep(100);
            } else {
                System.out.println("done");
                return true;
            }
        }
    }
    public static String getFileSHA512(File file) {
        String str = "";
        try {
            str = getHash(file, "SHA-512");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return str;
    }
    public static String getHash(File file, String hashType) throws Exception {
        InputStream fis = new FileInputStream(file);
        byte buffer[] = new byte[1024];
        MessageDigest md5 = MessageDigest.getInstance(hashType);
        for (int numRead = 0; (numRead = fis.read(buffer)) > 0; ) {
            md5.update(buffer, 0, numRead);
        }
        fis.close();
        return toHexString(md5.digest());
    }
    public static String toHexString(byte b[]) {
        StringBuilder sb = new StringBuilder();
        for (byte aB : b) {
            sb.append(Integer.toHexString(aB & 0xFF));
        }
        return sb.toString();
    }
    /**
     * 向文件写入byte[]
     *
     * @param fileName 文件名
     * @param bytes    字节内容
     * @param append   是否追加
     * @throws IOException
     */
    public static void writeFileByBytes(String fileName, byte[] bytes, boolean append) throws IOException {
        try(OutputStream out = new BufferedOutputStream(new FileOutputStream(fileName, append))){
            out.write(bytes);
        }
    }
}



